package com.rui.study.rocketmq;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static com.rui.study.rocketmq.Constant.*;

/**
 * @program: study_20180824
 * @description: 启动类
 * @author: Yaowr
 * @create: 2018-08-30 10:03
 **/
public class MessageSender {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageSender.class);


    public static void main(String[] args) {
//        String[] array = new String[5];
//        Arrays.fill(array, "hehe");
//        LOGGER.info(array.toString());
//        List<String> list = new ArrayList<String>(Collections.nCopies(3, "hello"));
//        LOGGER.info(list.toString());
//
//        Collections.fill(list, "world");
//        LOGGER.info(list.toString());

        syncSend();
    }

    public static void syncSend() {
        MQProducer producer = getProducer();
        try {
            for (int i = 1; i <= 1000; i++) {
                Message message = new Message(MQ_TOPIC, MQ_TAG, ("你好，世界 -- " + i).getBytes("utf-8"));
                SendResult result = producer.send(message);
                LOGGER.info("发送成功！ {}", result.toString());
            }
        } catch (Exception e) {
            LOGGER.error("Sending failed", e);
        } finally {
            producer.shutdown();
        }
    }

    public static void asyncSend() {
        MQProducer producer = getProducer();
        try {
            for (int i = 1; i <= 1000; i++) {
                Message message = new Message(MQ_TOPIC, MQ_TAG, ("你好，世界 -- " + i).getBytes("utf-8"));
                 producer.send(message, new SendCallback() {
                    public void onSuccess(SendResult result) {
                        LOGGER.info("发送成功！ {}", result.toString());
                    }

                    public void onException(Throwable e) {
                        throw new RuntimeException(e);
                    }
                });
            }
        } catch (Exception e) {
            LOGGER.error("Sending failed", e);
        } finally {
            producer.shutdown();
        }
    }


    private static MQProducer getProducer() {
        DefaultMQProducer producer = new DefaultMQProducer(Constant.MQ_PRODUCER_GROUP);
        producer.setNamesrvAddr(MQ_NAMESRV_ADDRESS);
        try {
            producer.start();
        } catch (MQClientException e) {
            LOGGER.error("Producer starting failed", e);
        }
        return producer;
    }
}
